package com.cdream.basic.mq.kafka;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @author cdream
 * @date 2022/10/26
 */
@Slf4j
public class DogDeserializer implements Deserializer<Dog> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public Dog deserialize(String s, byte[] bytes) {
        if (bytes == null) {
            return null;
        }
        if (bytes.length < 8) {
            log.info("太短了");
            return null;
        }
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        int ageLen, nameLen;
        String name;
        int age;
        nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);
        name = new String(nameBytes);

        ageLen = buffer.getInt();
        age = buffer.getInt();

        Dog dog = new Dog();
        dog.setAge(age);
        dog.setName(name);
        return dog;
    }


    @Override
    public void close() {

    }
}
